package com.fulihui.wgj.consumer.listener;

import com.alibaba.dubbo.config.annotation.Reference;
import com.fulihui.common.constants.TopicConstants;
import com.fulihui.common.enums.SendLogStatusEnum;
import com.fulihui.wgj.consumer.entity.CustomerNewsMsg;
import com.fulihui.wgj.consumer.result.BatchSendResult;
import com.fulihui.wgj.consumer.service.WxMsgService;
import com.fulihui.wgj.facade.msg.Wx48HMsg;
import com.fulihui.wgj.facade.request.wx48.Wx48NewsLogUpdateAccumRequest;
import com.fulihui.wgj.facade.service.wx48.Wx48NewsInfoService;
import org.apache.kafka.clients.consumer.ConsumerRecord;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.kafka.annotation.KafkaListener;
import org.springframework.stereotype.Component;

import java.util.Date;

/**
 * @author: LiuJie
 * @version: 2018/7/6 13:40
 * @description: 接收并发送 wx 48H Msg
 */
@Component
public class Wx48HMsgListener extends AbstractListener<Wx48HMsg> {
    @Autowired
    private WxMsgService wxMsgService;
    @Reference(version = "1.0.0")
    private Wx48NewsInfoService wx48NewsInfoService;

    @Override
    @KafkaListener(topics = TopicConstants.WX_48H_MSG)
    void listener(ConsumerRecord<String, String> record) {
        Wx48HMsg wx48HMsg = super.unpack(record.value(), Wx48HMsg.class);
        //发送客服48H自定义图文消息(news)
        CustomerNewsMsg customerNewsMsg = new CustomerNewsMsg(
                wx48HMsg.getAppId(),
                wx48HMsg.getNewsKey(),
                wx48HMsg.getOpenIds()
        );
        BatchSendResult result = wxMsgService.sendCustomerNewsMsg(customerNewsMsg);

        Long recordId = wx48HMsg.getRecordId();
        // 如果包含批次信息则更新批次信息。
        if(null != recordId){
            Wx48NewsLogUpdateAccumRequest request = new Wx48NewsLogUpdateAccumRequest();
            request.setId(wx48HMsg.getRecordId());
            request.setSendSuccessCount(Long.valueOf(result.getSuccessNum()));
            request.setSendFailCount(Long.valueOf(result.getFailNum()));
            // 最后一批次更改发送状态，but 因为kafka 消息 partition 间无序，所以未必是结束的标志，可能其他分区还未消费完
            if(wx48HMsg.isLast()){
                request.setStatus(SendLogStatusEnum.complete.getCode());
                request.setSendEndTime(new Date());
            }
            wx48NewsInfoService.updateNewsLogAccumulate(request);
        }
    }
}
